#include "vas_zk.h"
#include "vas_config.h"

namespace vas {

    zhandle_t *ZookeeperInit(const std::string &host, bool *running, const char *channel_root) {
        zhandle_t *zh = zookeeper_init(
                host.c_str(),
                nullptr,
                30000,
                nullptr,
                nullptr,
                0);
        if (zh == nullptr) {
            std::stringstream msg;
            msg << "Zookeeper init failed:";
            VAS_ERROR(msg.str());
            return nullptr;
        }
        std::shared_ptr<Stat> root_stat(new Stat());
        int res = zoo_wexists(zh, channel_root, ChannelRootWatcher, &running, root_stat.get());
        if (res != ZOK) {
            std::stringstream msg;
            msg << "The root path is not exist or readable, ret code:" << res;
            VAS_ERROR(msg.str());
            return nullptr;
        }
        return zh;
    }

    void ChannelRootWatcher(zhandle_t *zh, int type, int state, const char *path, void *watcherCtx) {
        std::stringstream info;
        info << "zk node " << path << " get event, type " << type <<" state " << state;
        VAS_INFO(info.str());
        if (type == ZOO_DELETED_EVENT) {
            std::stringstream msg;
            msg << "channel root node " << path << " is deleted";
            VAS_INFO(msg.str());
            *(bool *) watcherCtx = false;
        }
        else{
            std::shared_ptr<Stat> root_stat(new Stat());
            int res = zoo_wexists(zh, path, ChannelRootWatcher, watcherCtx, root_stat.get());
            if (res != ZOK) {
                std::stringstream msg;
                msg << "Re add root watcher inside watcher failed , ret code:" << res;
                VAS_ERROR(msg.str());
            }
        }

    }

    void ChannelHandleWatcher(zhandle_t *zh, int type, int state, const char *path, void *watcherCtx) {
        std::stringstream info;
        info << "zk node " << path << " get event, type " << type <<" state " << state;
        VAS_INFO(info.str());
        if (type == ZOO_DELETED_EVENT) {
            std::stringstream msg;
            msg << "Channel handle node " << path << " is deleted";
            VAS_INFO(msg.str());
            (*(void (*)(const std::string &channel_id)) watcherCtx)(
                    GetChannelFromHandlePath(path, CHANNEL_ROOT, HANDLE_NODE));
        }
        else{
            std::shared_ptr<Stat> root_stat(new Stat());
            int res = zoo_wexists(zh, path, ChannelHandleWatcher, watcherCtx, root_stat.get());
            if (res != ZOK) {
                std::stringstream msg;
                msg << "Re add Handle " << path << " Watcher inside Watcher failed , ret code:" << res;
                VAS_ERROR(msg.str());
            }
        }
    }

    Channel
    GetOneWaitingChannel(zhandle_t *zh, const char *channel_root, const char *handle_node, void *handle_watch_ctx) {
        std::shared_ptr<String_vector> channel_nodes(new String_vector{});
        int get_channel_ret = zoo_get_children(zh, channel_root, 0, channel_nodes.get());
        if (get_channel_ret != ZOK) {
            std::stringstream msg;
            msg << "Get children from channel root failed, ret " << get_channel_ret << " root " << channel_root;
            throw VasRuntimeError(msg.str());
        }

        for (int i = 0; i < channel_nodes->count; i++) {
            std::stringstream channel_path;
            channel_path << channel_root << "/" << channel_nodes->data[i];
            std::shared_ptr<String_vector> handle_info(new String_vector{});
            int get_handle_ret = zoo_get_children(zh, channel_path.str().c_str(), 0, handle_info.get());
            if (get_handle_ret != ZOK) {
                std::stringstream msg;
                msg << "Get children from channel failed, ret " << get_channel_ret << " channel " << channel_path.str();
                VAS_ERROR(msg.str());
                continue;
            }

            if (handle_info->count > 0) {
                std::stringstream msg;
                msg << "Channel " << channel_path.str() << " is being handled";
                VAS_INFO(msg.str());
                continue;
            }

            int channel_info_buf_len = 10000;
            std::shared_ptr<char> channel_info_buf(new char[channel_info_buf_len]);
            int get_channel_info_ret = zoo_get(
                    zh,
                    channel_path.str().c_str(),
                    0,
                    channel_info_buf.get(),
                    &channel_info_buf_len,
                    nullptr);

            if (get_channel_info_ret != ZOK) {
                std::stringstream msg;
                msg << "Channel " << channel_path.str() << " info get failed, ret " << get_channel_info_ret;
                VAS_INFO(msg.str());
                continue;
            }
            Channel channel;
            try {
                std::string channel_info = channel_info_buf.get();
                channel = JsonToChannel(StringToJson(channel_info));
            }
            catch (std::exception &e) {
                std::stringstream msg;
                msg << "Channel " << channel_path.str() << " info decode failed, info " << channel_info_buf.get();
                VAS_INFO(msg.str());
                continue;
            }
            std::stringstream handle_path;
            handle_path << channel_path.str() << "/" << handle_node;
            ACL ALL_ACL[] = {ZOO_PERM_ALL, ZOO_ANYONE_ID_UNSAFE};
            std::shared_ptr<ACL_vector> ALL_PERMS(new ACL_vector{1, ALL_ACL});
            int path_buffer_len = 100;
            std::shared_ptr<char> path_buffer(new char[path_buffer_len]);
            int create_handle_ret = zoo_create(
                    zh,
                    handle_path.str().c_str(),
                    nullptr,
                    -1,
                    ALL_PERMS.get(),
                    ZOO_EPHEMERAL,
                    path_buffer.get(),
                    path_buffer_len);
            if (create_handle_ret != ZOK) {
                std::stringstream msg;
                msg << "Create handle node  " << handle_path.str() << " failed, ret code " << create_handle_ret;
                VAS_INFO(msg.str());
                continue;
            }
            std::shared_ptr<Stat> handle_state(new Stat());
            int watch_handle_ret = zoo_wexists(zh, handle_path.str().c_str(), ChannelHandleWatcher, handle_watch_ctx,
                                               handle_state.get());
            if (watch_handle_ret != ZOK) {
                std::stringstream msg;
                msg << "Watch handle node  " << handle_path.str() << " failed, ret code " << create_handle_ret;
                VAS_INFO(msg.str());
            }
            return channel;
        }
        throw VasNotFound("No Waiting Chanel is found");
    }

    void ZookeeperStop(zhandle_t *zh) {
        zookeeper_close(zh);
    }

    std::string
    GetChannelFromHandlePath(const std::string &path, const std::string &channel_root, const std::string &handle_node) {
        std::string no_root_path = path.substr(channel_root.size(), path.size() - channel_root.size());
        return path.substr(channel_root.size(), path.size() - channel_root.size()).substr(1, no_root_path.size() -
                                                                                             handle_node.size() - 2);
    }

}